package io.kiki.stack.netty.channelManager;


import io.kiki.stack.netty.Constants;
import io.kiki.stack.netty.RpcConfigManager;
import io.kiki.stack.netty.config.*;
import io.kiki.stack.netty.exception.RemotingException;
import io.kiki.stack.netty.protocol.Codec;
import io.kiki.stack.netty.util.EventLoopGroupUtil;
import io.kiki.stack.netty.util.IoUtils;
import io.kiki.stack.netty.util.NamedThreadFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.flush.FlushConsolidationHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.Getter;
import lombok.NonNull;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import javax.net.ssl.KeyManagerFactory;
import javax.net.ssl.SSLEngine;
import javax.net.ssl.TrustManagerFactory;
import java.io.FileInputStream;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.security.KeyStore;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Setter
@Getter
public abstract class AbstractChannelFactory implements ChannelFactory {


    private static final EventLoopGroup EVENT_LOOP_GROUP = EventLoopGroupUtil.getEventLoopGroup(Runtime.getRuntime().availableProcessors() + 1, new NamedThreadFactory("netty-client-worker", true));

    private final Config config;
    private final Codec codec;
    private final ChannelHandler heartbeatHandler;
    private final ChannelHandler handler;

    protected Bootstrap bootstrap;

    public AbstractChannelFactory(@NonNull Codec codec, @NonNull ChannelHandler heartbeatHandler, @NonNull ChannelHandler handler, @NonNull Config config) {
        this.config = config;
        this.codec = codec;
        this.heartbeatHandler = heartbeatHandler;
        this.handler = handler;
    }

    @Override
    public void init(final ChannelEventHandler channelEventHandler) {
        bootstrap = new Bootstrap();
        bootstrap.group(EVENT_LOOP_GROUP).channel(EventLoopGroupUtil.getClientSocketChannelClass())//
                .option(ChannelOption.TCP_NODELAY, ConfigManager.tcp_nodelay())//
                .option(ChannelOption.SO_REUSEADDR, ConfigManager.tcp_so_reuseaddr())//
                .option(ChannelOption.SO_KEEPALIVE, ConfigManager.tcp_so_keepalive())//
                .option(ChannelOption.SO_SNDBUF, ConfigManager.tcp_so_sndbuf())//
                .option(ChannelOption.SO_RCVBUF, ConfigManager.tcp_so_rcvbuf());//

        // init netty write buffer water mark
        initWriteBufferWaterMark();

        // init byte buf allocator
        if (ConfigManager.netty_buffer_pooled()) {
            this.bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        } else {
            this.bootstrap.option(ChannelOption.ALLOCATOR, UnpooledByteBufAllocator.DEFAULT);
        }

        final boolean flushConsolidationSwitch = this.config.option(ClientOption.netty_flush_consolidation);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            //
            @Override
            protected void initChannel(SocketChannel socketChannel) {
                ChannelPipeline channelPipeline = socketChannel.pipeline();
                ExtendedChannelHandlerList extendedChannelHandlerList = config.option(ClientOption.extended_netty_channel_handler);
                if (extendedChannelHandlerList != null) {
                    List<ChannelHandler> list = extendedChannelHandlerList.frontChannelHandlers();
                    if (list != null) {
                        for (ChannelHandler channelHandler : list) {
                            channelPipeline.addLast(channelHandler.getClass().getName(), channelHandler);
                        }
                    }
                }
                Boolean sslEnable = config.option(ClientOption.cli_ssl_enable);
                if (!sslEnable) {
                    // fixme: remove in next version
                    sslEnable = RpcConfigManager.client_ssl_enable();
                }
                if (sslEnable) {
                    SSLEngine sslEngine = initSslContext().newEngine(socketChannel.alloc());
                    sslEngine.setUseClientMode(true);
                    channelPipeline.addLast(Constants.ssl_handler, new SslHandler(sslEngine));
                }
                if (flushConsolidationSwitch) {
                    channelPipeline.addLast("flushConsolidationHandler", new FlushConsolidationHandler(1024, true));
                }
                channelPipeline.addLast("decoder", codec.decoder());
                channelPipeline.addLast("encoder", codec.encoder());

                boolean idleSwitch = ConfigManager.tcp_idle_switch();
                if (idleSwitch) {
                    channelPipeline.addLast("idleStateHandler", new IdleStateHandler(ConfigManager.tcp_idle(), ConfigManager.tcp_idle(), 0, TimeUnit.MILLISECONDS));
                    channelPipeline.addLast("heartbeatHandler", heartbeatHandler);
                }
                channelPipeline.addLast("channelEventHandler", channelEventHandler);
                channelPipeline.addLast("handler", handler);
                if (extendedChannelHandlerList != null) {
                    List<ChannelHandler> list = extendedChannelHandlerList.backChannelHandlers();
                    if (list != null) {
                        for (ChannelHandler channelHandler : list) {
                            channelPipeline.addLast(channelHandler.getClass().getName(), channelHandler);
                        }
                    }
                }
            }
        });
    }

    @Override
    public Channel createChannel(Url url) throws Exception {
        io.netty.channel.Channel channel = doCreateChannel(url.getIp(), url.getPort(), url.getConnectTimeout());
        if (channel.isActive()) {
            channel.pipeline().fireUserEventTriggered(ChannelEventType.CONNECT);
        } else {
            channel.pipeline().fireUserEventTriggered(ChannelEventType.CONNECT_FAILED);
            throw new RemotingException("create channel, but channel is inactive, url is " + url.getOriginUrl());
        }
        return new Channel(channel, url);
    }

    @Override
    public Channel createChannel(String targetIP, int targetPort, int connectTimeout) throws Exception {
        io.netty.channel.Channel channel = doCreateChannel(targetIP, targetPort, connectTimeout);
        if (channel.isActive()) {
            channel.pipeline().fireUserEventTriggered(ChannelEventType.CONNECT);
        } else {
            channel.pipeline().fireUserEventTriggered(ChannelEventType.CONNECT_FAILED);
            throw new RemotingException("create channel, but channel is inactive, target address is " + targetIP + ":" + targetPort);
        }
        return new Channel(channel, new Url(targetIP, targetPort));
    }

    /**
     * init netty write buffer water mark
     */
    private void initWriteBufferWaterMark() {
        // init with system properties
        Integer lowWatermark = ConfigManager.netty_buffer_low_watermark();
        if (lowWatermark != null) {
            config.option(ServerOption.netty_buffer_low_water_mark, lowWatermark);
        }
        Integer highWatermark = ConfigManager.netty_buffer_high_watermark();
        if (highWatermark != null) {
            config.option(ServerOption.netty_buffer_high_water_mark, highWatermark);
        }

        int lowWaterMark = config.option(GenericOption.netty_buffer_low_water_mark);
        int highWaterMark = config.option(GenericOption.netty_buffer_high_water_mark);
        if (lowWaterMark > highWaterMark) {
            throw new IllegalArgumentException(String.format("[client side]  netty high water mark {%s} should not be smaller than low water mark {%s} bytes)", highWaterMark, lowWaterMark));
        } else {
            log.warn("[client side]  netty low water mark is {} bytes, high water mark is {} bytes", lowWaterMark, highWaterMark);
        }
        this.bootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(lowWaterMark, highWaterMark));
    }

    private SslContext initSslContext() {
        InputStream inputStream = null;
        try {
            String sslKeyStoreType = config.option(ClientOption.cli_ssl_keystore_type);
            if (sslKeyStoreType == null) {
                // fixme: remove in next version
                sslKeyStoreType = RpcConfigManager.client_ssl_keystore_type();
            }
            KeyStore keyStore = KeyStore.getInstance(sslKeyStoreType);
            String sslKeyStore = config.option(ClientOption.cli_ssl_keystore);
            if (sslKeyStore == null) {
                sslKeyStore = RpcConfigManager.client_ssl_keystore();
            }
            inputStream = new FileInputStream(sslKeyStore);
            String keyStorePass = config.option(ClientOption.cli_ssl_keystore_pass);
            if (keyStorePass == null) {
                keyStorePass = RpcConfigManager.client_ssl_keystore_pass();
            }
            char[] keyStorePassBytes = keyStorePass.toCharArray();
            keyStore.load(inputStream, keyStorePassBytes);
            String serverSslAlgorithm = config.option(ServerOption.SRV_SSL_KMF_ALGO);
            if (serverSslAlgorithm == null) {
                serverSslAlgorithm = RpcConfigManager.server_ssl_kmf_algorithm();
            }
            KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(serverSslAlgorithm);
            keyManagerFactory.init(keyStore, keyStorePassBytes);
            String sslAlgorithm = config.option(ClientOption.cli_ssl_tmf_algo);
            if (sslAlgorithm == null) {
                sslAlgorithm = RpcConfigManager.client_ssl_tmf_algorithm();
            }
            TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(sslAlgorithm);
            trustManagerFactory.init(keyStore);
            return SslContextBuilder.forClient().keyManager(keyManagerFactory).trustManager(trustManagerFactory).build();
        } catch (Exception e) {
            log.error("Fail to init SSL context", e);
            throw new IllegalStateException("Fail to init SSL context", e);
        } finally {
            IoUtils.closeQuietly(inputStream);
        }

    }

    protected io.netty.channel.Channel doCreateChannel(String targetIP, int targetPort, int connectTimeout) throws Exception {
        String address = targetIP + ":" + targetPort;
        if (connectTimeout <= 0) {
            throw new IllegalArgumentException(String.format("illegal timeout for creating channel, address: %s, timeout: %d", address, connectTimeout));
        }
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout);
        ChannelFuture channelFuture = bootstrap.connect(new InetSocketAddress(targetIP, targetPort));
        channelFuture.awaitUninterruptibly();
        if (!channelFuture.isDone()) {
            String errMsg = "Create channel to " + address + " timeout!";
            log.warn(errMsg);
            throw new Exception(errMsg);
        }
        if (channelFuture.isCancelled()) {
            String errMsg = "Create channel to " + address + " cancelled by user!";
            log.warn(errMsg);
            throw new Exception(errMsg);
        }
        if (!channelFuture.isSuccess()) {
            String errMsg = "Create channel to " + address + " error!";
            log.warn(errMsg);
            throw new Exception(errMsg, channelFuture.cause());
        }
        return channelFuture.channel();
    }
}
